package cn.witsky.rtp2ws.domain;

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicStampedReference;
import java.util.concurrent.locks.LockSupport;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.witsky.rtp2ws.Server;
import cn.witsky.rtp2ws.cache.ClientManager;
import cn.witsky.rtp2ws.client.ws.Client;
import cn.witsky.rtp2ws.client.ws.ClientListener;
import cn.witsky.rtp2ws.client.ws.ClientUtil;
import cn.witsky.rtp2ws.push.disruptor.EventCacheManage;
import cn.witsky.rtp2ws.push.threadpool.PushManager;
import cn.witsky.rtp2ws.push.threadpool.PushTask;
import cn.witsky.rtp2ws.util.FileUtil;
import cn.witsky.rtp2ws.util.LogUtil;
import cn.witsky.rtp2ws.util.SystemClock;

public class Rtp {

	private static Logger logger = LoggerFactory.getLogger(Rtp.class);
	// 标识Id
	private String rtpId;
	// 是否是主叫
	private boolean isCaller;
	// 呼叫基本信息
	private CallInfo call = null;
	// rtp流原地址
	private String source = "";
	// rtp流缓存
	RtpCached rtpCache =null;
	// websocket连接
	private volatile Client client = null;
	// 发送最大延迟
	private int sendWsMaxDelayTime = 0;
	// 发送延迟次数
	private int sendWsDelayCount = 0;
	// 发送ws最大延迟的seq
	private int sendWsDelaySeq = 0;
	// 当前发送数据量
	private volatile int sendSeq = 0;
	// 发送次数
	private int sendCount;
	// 发送成功次数
	private AtomicInteger sendSucCount = new AtomicInteger();
	// 发送失败次数
	private AtomicInteger sendErrCount = new AtomicInteger();
	// 发送结束时间
	private volatile long finishTime = -1l;
	// 重试记录
	private Retry retry = new Retry();
	// 会话结束标记
	private volatile boolean finished = false;
	// 读取空包合计次数
	private int readNullSumCount;
	// 连续读取空包最大次数
	private int readNullMaxCount;
	// 是否延迟
	private boolean delay = false;
	// 消息推送Task
	private volatile PushTask task;
	// websocket监听器
	private ClientListener clientlistener = null;
	// 是否是retry
	private boolean retransmission = false;

	private long lastSendTime = 0l;

	public Rtp(CallInfo call, boolean isCaller) {
		this.isCaller = isCaller;
		this.call = call;
		String callType = "";
		if (this.isCaller) {
			callType = "caller";
		} else {
			callType = "called";
		}
		this.rtpId = this.call.getCallId() + "-" + callType;
		clientlistener = new ClientListener(this);
		rtpCache=new RtpCached(this.rtpId);
	}
	public boolean getTaskStatus() {
		if(task==null) {
			return true;
		}else {
			return task.isFinished();
		}
	}
	// 初始化client
	public int initWsClient() {
		logger.debug("task:{} client:{}",this.getRtpId(),this.client);
		if (client == null) {
			long begin = SystemClock.now();
			synchronized (this) {
				if (client != null) {
					return 0;
				}
				Client wsClient = ClientUtil.getClient(this, retransmission);
				if (wsClient.isValid()) {
					client = wsClient;
					retransmission = true;
					client.addListener(clientlistener);
					this.client.setSubscribe(true);
					try {
						this.client.connectBlocking();
					} catch (InterruptedException e) {
						LogUtil.warn(this, "Rtp[initWsClient:105]", "task:{}创建的client是无效的", this.getRtpId(), client);
					}
				} else {
					LogUtil.info(this, "Rtp[initWsClient:108]", "task:{}创建的client是无效的", this.getRtpId(), client);
					return -1;
				}
			}
			return (int) (SystemClock.now() - begin);
		} else {
			return 0;
		}
	}
	
	// 清理wsClient
	public void clearWsClient() {
		int byteVersion = retry.incrCount();
		clientlistener.setByteVersion(byteVersion);
		client.setSubscribe(false);
		client = null;
		if(!this.finished) {
			this.creatTask();
		}

		LogUtil.info(this, "Rtp[clearWsClient:125]", "task:{} 清理client:{}", this.getRtpId(), this.client);
	}
	
	public int sendMsg(int retryCount, int version) {
		int sendSize = 0;
		if (!this.retry.isStart()) {
			boolean suc = this.retry.start(retryCount, version);
			if (suc) {
				int reciveBytes = this.getRetry().getReciveByte();
				if (reciveBytes == -1) {
					if (this.finishTime > 0) {
						this.finished();
					}
				} else {
					int seq = this.findSeq(reciveBytes);
					this.sendSeq=seq;
					logger.info("task:{} 重传seq:{}",this.rtpId,seq);
					Message msg = rtpCache.getMessage(seq);
					if (msg == null) {
						logger.warn("task:{}没有读到数据", this.rtpId);
						sendSize = -1;
					} else {
						logger.warn("task:{}查找到序列号:{}", this.getRtpId(), seq);
						sendSize = this.getBatch(msg.getBytes());
						suc = this.sendBytes();
						if (!suc) {
							logger.warn("task:{}没有读到数据", this.rtpId);
							sendSize = -1;
							if (this.finishTime > 0) {
								this.finished();
							}
						}
					}

				}
				return sendSize;
			} else {
				logger.warn("task:{}被终止了，需要重新发送", this.rtpId);
				return -1;
			}

		} else {
			sendSize = this.getBatch(null);
			logger.debug("task:{} sendSize:{} sendSeq:{}", this.rtpId, sendSize, this.sendSeq);
			boolean suc = this.sendBytes();
			if (!suc) {
				LogUtil.warn(this, "Rtp[sendMsg:171]", "task:{}没有读到数据", this.getRtpId());
				sendSize = -1;
				if (this.finishTime > 0) {
					this.finished();
				}
			}
			return sendSize;
		}
	}
	
	public String getCallId() {
		return this.call.getCallId();
	}

	public String getRtpId() {
		return rtpId;
	}

	public String getWsUrl() {
		if (this.isCaller) {
			return call.getCallerWsUrl();
		}
		return call.getCalledWsUrl();
	}

//	public String getCallUser() {
//		if (this.isCaller) {
//			return call.getCallId() + "_" + call.getCaller();
//		}
//		return call.getCallId() + "_" + call.getCalled();
//	}
	
	public String getCallUser() {
		if (this.isCaller) {
			return call.getCallId() + "_1";
		}
		return call.getCallId() + "_2";
	}

	public void setTask(PushTask task) {
		this.task = task;
	}

	public void setSource(String source) {
		this.source = source;
	}

	public boolean isCaller() {
		return isCaller;
	}

	public AtomicInteger getSendSucCount() {
		return sendSucCount;
	}

	public AtomicInteger getSendErrCount() {
		return sendErrCount;
	}

	public boolean isFinished() {
		return finished;
	}

	public void incrSendSeq() {
		this.sendSeq++;
	}

	public void setReadNullSumCount() {
		this.readNullSumCount++;
	}

	public void setReadNullMaxCount(int readNullMaxCount) {
		if (readNullMaxCount > this.readNullMaxCount) {
			this.readNullMaxCount = readNullMaxCount;
		}

	}

	// 时间大于140毫秒,记录延迟次数
	public void setSendWsMaxDelayTime(int delayTime) {
		if (delayTime > 140) {
			sendWsDelayCount++;
		}
		if (delayTime > this.sendWsMaxDelayTime) {
			this.sendWsMaxDelayTime = delayTime;
			sendWsDelaySeq = this.sendCount;
		}
	}

	public Retry getRetry() {
		return retry;
	}



	public Client getClient() {
		return this.client;
	}

	public boolean clientIsNull() {
		if (client == null) {
			return true;
		} else {
			return false;
		}
	}



	// 关闭client
	public void clientClose() {
		if (client != null) {
			this.client.setSubscribe(false);
			this.client.close();
		}

	}
	public void creatTask() {
		if(task==null) {
			synchronized (this) {
				if(task==null) {
					task = new PushTask(this);
					PushManager.execTask(task);
				}

			}

		}
	}
	public boolean hasTask() {
		return task!=null;
	}
	// 唤醒线程
	public void wakeUpTask() {

		LockSupport.unpark(task.getCurrentThread());
	}

	// 接收rtp流
	public void offerRtpDataSync(int seq, byte[] bytes) {
		logger.debug("task:{} seq:{} bytes:{}", this.getRtpId(), seq, bytes.length);
		call.setReciveByteTime();
		Message message = new Message(this.isCaller());
		message.setSeq(seq);
		message.setBytes(bytes);
		message.setCallId(this.getCallId());
		this.sendMsg(message);
	}
	
	// 接收rtp流
	public void offerRtpData(int seq, byte[] bytes) {
		logger.debug("task:{} seq:{} bytes:{}", this.getRtpId(), seq, bytes.length);
		call.setReciveByteTime();
		if (Server.useNewThread) {
			Message message = new Message(this.isCaller());
			message.setSeq(seq);
			message.setBytes(bytes);
			message.setCallId(this.getCallId());
			this.rtpCache.addMessage(message);
		} else {
			EventCacheManage.offer(isCaller, this.getCallId(), seq, bytes, SystemClock.now());
		}
	}

	// 通过响应的reciveBytes查找需要从什么位置开始发包
	public int findSeq(int reciveBytes) {
		int start = 1;
		if (reciveBytes == 0) {
			return start;
		} else {
			int seq = reciveBytes / Server.getPayLoadLen();
			return seq + 1;
		}
	}

	public void incrSendCount() {
		this.sendCount++;
	}

	public long getFinishTime() {
		return finishTime;
	}

	public void setFinishTime(Long finishTime) {
		this.finishTime = finishTime;
		if (this.task != null) {
			LockSupport.unpark(this.task.getCurrentThread());
			logger.debug("task:{}唤醒了线程", this.getRtpId());
		} else {
			if(Server.sendWsSync) {
				Message message=new Message();
				message.setCallId(this.getCallId());
				message.setBytes(new byte[0]);
				message.setCaller(this.isCaller);
				message.setSeq(-1);
				message.setCurrentTimeMillis(finishTime);
				this.sendMsg(message);
			}else {
				EventCacheManage.offer(isCaller, this.getCallId(), -1, new byte[0], finishTime);
			}
			logger.debug("task:{}通知消费者会话结束了", this.getRtpId());
		}
	}

	public Monitor getMonitorData() {
		Monitor monitor = new Monitor(call.getCallId(), this.isCaller);
		monitor.setSource(this.source);
		for (int i = 1; i <= rtpCache.size(); i++) {
			Message message = rtpCache.getMessage(i);
			if (message == null) {
				break;
			}
			monitor.setCurrentTime(message.getCurrentTimeMillis(), message.getSeq());
		}
		monitor.setMsgCount(rtpCache.size());
		monitor.setAvg();
		monitor.setSendMsgCount(this.sendSeq);
		monitor.setSendCount(this.sendCount);
		monitor.setSendSucCount(this.getSendSucCount().get());
		monitor.setSendErrCount(this.getSendErrCount().get());
		monitor.setSendMaxdelayTime(this.sendWsMaxDelayTime);
		monitor.setSendWsdelayCount(this.sendWsDelayCount);
		monitor.setSendMaxdelaySeq(this.sendWsDelaySeq);
		monitor.setReadNullMaxCount(readNullMaxCount);
		monitor.setReadNullSumCount(readNullSumCount);
		if(this.isCaller) {
			monitor.setDstPort(call.getUdpPort());
		}else {
			monitor.setDstPort(call.getUdpPort()+1);
		}
		if(Server.isSaveFile()) {
			this.backUpFile();
		}
		return monitor;
	}
	public String getPcmName() {
		return this.rtpId+".pcm";
	}
	public String getPcmTarName() {
		return this.rtpId+".pcm.tar.gz";
	}
	private void backUpFile() {
		if(rtpCache.size()==0) {
			return;
		}
		String filePath=Server.backUpRtpFileDir+"/"+this.getPcmName();		
		if(call.hasPcm()) {
			logger.info("task:{} 文件{}已经存在了",this.getRtpId(),filePath);
			return ;
		}
		logger.info("task:{} 开始备份文件{}",this.getRtpId(),filePath);
		File file = new File(filePath);    //1、建立连接
		DataOutputStream d =null;
		try {
			file.createNewFile();
			d=new DataOutputStream(new FileOutputStream(file));
			for (int i = 1; i <= rtpCache.size(); i++) {
				Message message = rtpCache.getMessage(i);
				if (message == null) {
					break;
				}
				if(message.getSeq()>-1&&message.getBytes()!=null) {
					d.write(message.getBytes());
					d.flush();
				}
			}
			logger.info("task:{} 备份文件{}成功",this.getRtpId(),filePath);
			if(Server.isCompress()) {
				boolean isSuc=FileUtil.tar(Server.backUpRtpFileDir,this.getPcmName(),Server.backUpRtpFileDir,this.getPcmTarName());
				if(isSuc) {
					file.delete();
				}
			}
		}catch(Exception ex) {
			logger.error("task:"+this.getRtpId()+"备份文件"+filePath+"异常:",ex);
		}finally {
			if(d!=null) {
				try {
					d.close();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}
	
	public int getSendSeq() {
		return sendSeq;
	}

	// 根据要求1s最多100个数据包,10毫秒就是1个数据包，根据发送的频率计算需要休眠的时间
	private int getSendSize(boolean isAll) {
		int restCount = rtpCache.getRtpDataSize() - this.sendSeq;
		if(isAll) {			
			return restCount;
		}else {
			logger.debug("task:{} rtpSize:{} sendSeq:{}", this.getRtpId(), rtpCache.getRtpDataSize(), sendSeq);
			int maxCount = (int) (Server.getWaiTime() / 2);
			if (restCount > maxCount) {
				return maxCount;
			} else {
				return restCount;
			}
		}	
	}

	// 批量将数据读取到buf中
	private int getBatch(byte[] bytes) {
		int sendSize = 0;
		int batchSize = this.getSendSize(false);
		if (bytes != null) {
			batchSize = batchSize - 1;
			rtpCache.bufWrite(this.getRtpId(),bytes);
			sendSize++;
		}
		logger.debug("task:{} getBatchSize:{}", this.getRtpId(), batchSize);
		if (batchSize == 0) {
			return sendSize;
		}
		sendSize = this.readMessage(batchSize);
		return sendSize;
	}

	public void getBatchNew(boolean isAll) {
		int batchSize = this.getSendSize(isAll);
		if (batchSize == 0) {
			return;
		}
		logger.debug("tasK:{},batchSize:{},delay:{} finishTime:{} batchSize>= Server.getOnceSendBytelen():{}",
				this.getRtpId(), batchSize, delay, finishTime, batchSize >= Server.getOnceSendBytelen());
		if (this.finishTime < 0) {
			if (delay) {
				this.readMessage(batchSize);
			} else {
				if (batchSize >= Server.getOnceSendBytelen()) {
					this.readMessage(Server.getOnceSendBytelen());
				} else {
					// 没有达到要发送的指标不发送
				}
			}
		} else {
			this.readMessage(batchSize);
		}
	}

	public int readMessage(int batchSize) {
		int sendSize = 0;
		Message msg = null;
		logger.debug("task:{} batchSize:{}", this.rtpId, batchSize);
		for (int i = 0; i < batchSize; i++) {
			msg = rtpCache.getMessage(this.sendSeq + 1);
			if (msg == null) {
				break;
			}
			rtpCache.bufWrite(this.getRtpId(),msg.getBytes());
			this.incrSendSeq();
			sendSize++;
		}
		return sendSize;
	}

	private boolean sendBytes() {
		try {
			if (this.client == null) {
				return false;
			}
			if (!this.client.isValid()) {
				return false;
			}
			if (this.client != null) {
				boolean result = client.sendBinary();
				if (!result) {
					return false;
				} else {
					incrSendCount();
					return true;
				}
			} else {
				return false;
			}
		}catch(Exception ex) {
			return false;
		}


	}

	public void finished() {
		logger.debug("task:{}停止rtp流推送...........", this.getRtpId());
		if (!finished) {
			finished = true;
			this.call.getRtpCount().decrementAndGet();
		}
	}

//	public void sendMsg(Message message) {
//		if (!this.hasTask()) {
//			this.initWsClient();
//			if (this.clientIsNull()) {
//				this.creatTask();
//			}
//		}
//		if (this.hasTask()) {
//			if (message.getSeq() > -1) {
//				rtpCache.addMessage(message);
//			} else {
//				this.setFinishTime(message.getCurrentTimeMillis());
//			}
//		} else {
//			if (message.getSeq() > -1) {
//				rtpCache.addMessage(message);
//				int delayTime=(int) (SystemClock.now()-this.lastSendTime);
//				delay=this.isDelay(delayTime);
//				this.getBatchNew(false);
//				boolean suc = this.sendBytes();
//				logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc, false);
//			} else {
//				boolean suc = false;
//				while (true) {
//					this.getBatchNew(true);
//					int writIndex = this.rtpCache.getBuf().writerIndex();
//					if (writIndex == 0) {
//						break;
//					} else {
//						suc = this.sendBytes();
//					}
//					logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc,true);
//				}
//				suc =this.sendEndMsg();
//				logger.debug("task:{}发送结束标记了......", this.rtpId);
//				logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc, true);
//				this.finished();
//			}
//		}
//	}

//	public void sendMsg(Message message) {
//		if (!this.hasTask()) {
//			if(this.clientIsNull()) {
//				ClientManager.getClient(this);
//			}
//		}
//		if (this.hasTask()) {
//			if (message.getSeq() > -1) {
//				rtpCache.addMessage(message);
//			} else {
//				this.setFinishTime(message.getCurrentTimeMillis());
//			}
//		}else {
//			if(!this.clientIsNull()) {
//				if (message.getSeq() > -1) {
//					rtpCache.addMessage(message);
//					int delayTime=(int) (SystemClock.now()-this.lastSendTime);
//					delay=this.isDelay(delayTime);
//					this.getBatchNew(false);
//					boolean suc = this.sendBytes();
//					logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc, false);
//				} else {
//					boolean suc = false;
//					while (true) {
//						this.getBatchNew(true);
//						if(rtpCache.getByteSzie()==0) {
//							break;
//						} else {
//							suc = this.sendBytes();
//						}
//						logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc,true);
//					}
//					suc =this.sendEndMsg();
//					logger.debug("task:{}发送结束标记了......", this.rtpId);
//					logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc, true);
//					this.finished();
//				}
//			}else {
//				rtpCache.addMessage(message);
//			}
//		}
//	}

	public void sendMsg(Message message) {
		rtpCache.addMessage(message);
		if (!this.hasTask()) {
			if(this.clientIsNull()) {
				ClientManager.getClient(this);
			}
		}
		if (this.hasTask()) {
			if (message.getSeq()<0) {
				this.setFinishTime(message.getCurrentTimeMillis());
			}
		}else {
			if(!this.clientIsNull()) {
				if (message.getSeq() > -1) {
					int delayTime=(int) (SystemClock.now()-this.lastSendTime);
					delay=this.isDelay(delayTime);
					this.getBatchNew(false);
					boolean suc = this.sendBytes();
					logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc, false);
				} else {
					boolean suc = false;
					while (true) {
						this.getBatchNew(true);
						if(rtpCache.getByteSzie()==0) {
							break;
						} else {
							suc = this.sendBytes();
						}
						logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc,true);
					}
					suc =this.sendEndMsg();
					logger.debug("task:{}发送结束标记了......", this.rtpId);
					logger.debug("task:{} message seq:{} suc:{} isAll:{}", this.getRtpId(), message.getSeq(), suc, true);
					this.finished();
				}
			}
		}
	}
	
	public boolean sendEndMsg() {
		boolean isSuc=false;
//		initWsClient();
		if (client != null) {
			isSuc=client.sendEndFlag();
			logger.debug("task:{}发送结束标记了......", this.getRtpId());
			incrSendCount();
		}
		return isSuc;

	}

	public RtpCached getRtpCache() {
		return rtpCache;
	}

	public boolean isDelay(int delayTime) {
		logger.debug("task:{} delayTime:{} waitTime:{}", this.getRtpId(), delayTime, Server.getWaiTime());
		if (delayTime > Server.getWaiTime()) {
			return true;
		}
		return false;
	}

	public int getWaittime() {
		return Server.getWaiTime();
	}

	public long getLastSendTime() {
		return lastSendTime;
	}

	public void setLastSendTime() {
		long now=SystemClock.now();
		logger.debug("task:{} lastSendTime:{} nowSendTime:{} delayTime:{}",this.getRtpId(),this.lastSendTime,now,now-this.lastSendTime);
		int delayTime =0;
		if (this.lastSendTime >0) {
			delayTime=(int) (now - this.lastSendTime);			
		}
		delay = this.isDelay(delayTime);
		this.setSendWsMaxDelayTime(delayTime);
		this.lastSendTime = now;
	}

	public CallInfo getCallInfo() {
		return this.call;
	}
	public class Retry {
		AtomicInteger count = new AtomicInteger();
		AtomicStampedReference<Integer> reciveByte = new AtomicStampedReference<Integer>(0, 0);
		AtomicStampedReference<Boolean> start = new AtomicStampedReference<Boolean>(true, 0);

		public synchronized int incrCount() {
			count.incrementAndGet();
			int byteVersion;
			while (true) {
				int version = start.getStamp();
				boolean suc = start.compareAndSet(start.getReference(), false, version, version + 1);
				if (suc) {
					break;
				}
			}
			while (true) {
				byteVersion = reciveByte.getStamp();
				boolean suc = reciveByte.compareAndSet(reciveByte.getReference(), -1, byteVersion, byteVersion + 1);
				if (suc) {
					break;
				}
			}
			return reciveByte.getStamp();
		}

		public boolean start(int retryCount, int version) {
			if (retryCount == this.getCount()) {
				boolean suc = start.compareAndSet(false, true, version, version + 1);
				return suc;
			} else {
				return false;
			}
		}

		public int getReciveByte() {
			return reciveByte.getReference();
		}

		public boolean setReciveByte(int bytesLen, int version) {
			logger.debug("task:{} currentVal:{} newVal:{} stamp:{} version:{}",getRtpId(),this.getReciveByte(),bytesLen,this.getReciveByteVersion(),version);
			return reciveByte.compareAndSet(-1, bytesLen, version, version + 1);
		}

		public int getCount() {
			return count.get();
		}

		public boolean isStart() {
			return start.getReference();
		}

		public int getVersion() {
			return start.getStamp();
		}

		public int getReciveByteVersion() {
			return reciveByte.getStamp();
		}
	}

}
